package com.hilin.etcd.mq;

import com.hilin.common.util.EncryptUtils;
import com.hilin.etcd.mq.config.HilinEtcdMqConfig;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import mousio.etcd4j.EtcdClient;
import mousio.etcd4j.responses.EtcdKeysResponse;
import org.springframework.util.ObjectUtils;

import java.net.URI;
import java.util.Map;
import java.util.concurrent.*;

@Slf4j
public class HilinEtcdMqClient {

    private static final String EK_LISTENER_ROOT = "/com/hilin/mq/listener/";

    private static final String EK_LISTENER_KEY = EK_LISTENER_ROOT + "%s";

    private static final Map<String, HilinEtcdMqListener> MQ_LISTENER = new ConcurrentHashMap<>();

    private static final long[] INDEX = new long[]{0};

    private static Thread THREAD_LISTENER;

    public static ThreadPoolExecutor POOL_EXECUTOR;

    public HilinEtcdMqClient(EtcdClient etcdClient, HilinEtcdMqConfig etcdConfig) {
        if (THREAD_LISTENER == null) {
            POOL_EXECUTOR = new ThreadPoolExecutor(etcdConfig.getIdleConsumerNum(), etcdConfig.getMaxConsumerNum(),
                    1, TimeUnit.MINUTES, new LinkedBlockingDeque<>(etcdConfig.getMaxQueueNum()));
            THREAD_LISTENER = new Thread(() -> {
                while (true) {
                    synchronized (INDEX) {
                        try {
                            EtcdKeysResponse response;
                            if (0 == INDEX[0]) {
                                response = etcdClient.get(EK_LISTENER_ROOT).waitForChange().recursive().timeout(1, TimeUnit.MINUTES).send().get();
                            } else {
                                response = etcdClient.get(EK_LISTENER_ROOT).waitForChange(INDEX[0]).recursive().timeout(1, TimeUnit.MINUTES).send().get();
                            }
                            INDEX[0] = response.node.modifiedIndex + 1;
                            POOL_EXECUTOR.execute(() -> {
                                HilinEtcdMqListener listener = MQ_LISTENER.get(response.node.key);
                                if (!ObjectUtils.isEmpty(listener)) {
                                    try {
                                        listener.onMessage(EncryptUtils.decryptAES(response.node.key.replace(EK_LISTENER_ROOT, ""), "1"), response.node.value, response.node.modifiedIndex);
                                    } catch (Exception e) {
                                        e.printStackTrace();
                                    }
                                }
                            });
                        } catch (TimeoutException ignored) {
                        } catch (Exception e) {
                            log.info("etcd队列轮询失败", e);
                        }
                    }
                }
            });
            THREAD_LISTENER.start();
        }
    }

    public void addListenKey(String key, HilinEtcdMqListener listener) {
        if (!ObjectUtils.isEmpty(key) && !ObjectUtils.isEmpty(listener)) {
            key = keyDeal(key);
            System.out.println(key);
            MQ_LISTENER.put(key, listener);
        }
    }

    public void delListenKey(String key) {
        if (!ObjectUtils.isEmpty(key)) {
            key = keyDeal(key);
            MQ_LISTENER.remove(key);
        }
    }

    @SneakyThrows
    private String keyDeal(String key) {
        return String.format(EK_LISTENER_KEY, EncryptUtils.encryptAES(key, "1"));
    }


    public static void main(String[] args) {
        EtcdClient etcdClient = new EtcdClient(URI.create("http://localhost:2379"));
        HilinEtcdMqClient client = new HilinEtcdMqClient(etcdClient, new HilinEtcdMqConfig());
        client.addListenKey("/test/1", (key, value, modifiedIndex) -> System.out.println(key + " -> " + value));
        client.addListenKey("/test/2", (key, value, modifiedIndex) -> System.out.println(key + " -> " + value));
    }


}
